package org.example.server.search.listener.mq.kafka;

import cn.hutool.core.lang.TypeReference;
import cn.hutool.json.JSONUtil;
import co.elastic.clients.elasticsearch.ElasticsearchClient;
import co.elastic.clients.elasticsearch._types.Conflicts;
import co.elastic.clients.elasticsearch._types.InlineScript;
import co.elastic.clients.elasticsearch._types.Script;
import co.elastic.clients.elasticsearch._types.query_dsl.Query;
import co.elastic.clients.elasticsearch.core.UpdateByQueryRequest;
import co.elastic.clients.json.JsonData;
import jakarta.annotation.Resource;
import lombok.SneakyThrows;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.example.api.goods.category.controller.CategoryControllerFeign;
import org.example.api.goods.category.response.CategorySummaryResponse;
import org.example.api.goods.goods.controller.GoodsControllerFeign;
import org.example.api.goods.goods.response.GoodsFatDetailResponse;
import org.example.frw.canal.EntityComposite;
import org.example.frw.canal.HashValue;
import org.example.frw.canal.binlog.BaseCanalBinlog;
import org.example.frw.canal.binlog.entity.GoodsBinlogEntity;
import org.example.frw.canal.config.TopicConcurrentlySetting;
import org.example.frw.canal.listener.mq.kafka.BaseConsumer;
import org.example.server.search.doc.GoodsDOC;
import org.example.server.search.doc.GoodsDOC.Category;
import org.example.server.search.mapstruct.GoodsDocConverter;
import org.example.server.search.repo.GoodsDOCRepo;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.listener.BatchAcknowledgingMessageListener;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.retry.annotation.Backoff;
import org.springframework.retry.annotation.Retryable;
import org.springframework.stereotype.Component;
import org.springframework.stereotype.Service;

import java.io.IOException;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

@Component
@Slf4j
public class GoodsConsumer extends BaseConsumer<GoodsBinlogEntity> implements BatchAcknowledgingMessageListener<String, String> {
    private static final String TOPIC = "goods__tb_goods";
    private static final String GROUP = "search-server";


    @Resource
    private GoodsDOCRepo goodsDOCRepo;

    @Resource
    private GoodsRetryService goodsRetryService;

    @Resource
    private GoodsControllerFeign goodsFeign;

    @Resource
    private CategoryControllerFeign categoryFeign;

    @Resource
    private GoodsDocConverter goodsDocConverter;

    public GoodsConsumer(TopicConcurrentlySetting topicConcurrentlySetting){
        super(topicConcurrentlySetting, TOPIC ,GROUP);
    }

    /**
     * 幂等由ES保证
     * @param entityComposite entityComposite
     */
    @Override
    @SneakyThrows
    protected void doProcess(EntityComposite<? extends HashValue> entityComposite) {
        log.info("GoodsConsumer processing index = {},entityComposite={}",entityComposite.getPartition(), entityComposite);
        GoodsBinlogEntity goodsEntity = (GoodsBinlogEntity) entityComposite.getEntity();
        switch (entityComposite.getType()) {
            case "UPDATE" -> {
                GoodsBinlogEntity oldGoodsEntity = (GoodsBinlogEntity) entityComposite.getOldEntity();
                String scriptSource = """
                            ctx._source.title =  params.title;
                            ctx._source.subTitle =  params.subTitle;
                            ctx._source.status =  params.status;
                            if(params.category != null){
                                ctx._source.category = params.category;
                            }
                            ctx._source.updateTime =  params.updateTime;
                        """;
                Map<String, JsonData> params = new HashMap<>();
                params.put("title", JsonData.of(goodsEntity.getTitle()));
                params.put("subTitle", JsonData.of(goodsEntity.getSubTitle()));
                params.put("status", JsonData.of(goodsEntity.getStatus()));
                params.put("updateTime", JsonData.of(goodsEntity.updateTimeTimestamp()));
                if (oldGoodsEntity.getCategoryId() != null) {
                    // 分类发生了变化
                    params.put("category", JsonData.of(category(goodsEntity.getCategoryId())));
                }
                goodsRetryService.doUpdateQuery(goodsEntity,scriptSource,params);
            }
            case "INSERT" -> {
                GoodsDOC goodsDOC = new GoodsDOC();

                GoodsFatDetailResponse fatDetailResponse = goodsFeign.fatDetail(goodsEntity.getId());
                goodsDOC.setId(fatDetailResponse.getId());
                goodsDOC.setTitle(fatDetailResponse.getTitle());
                goodsDOC.setSubTitle(fatDetailResponse.getSubTitle());
                goodsDOC.setStatus(fatDetailResponse.getStatus());
                goodsDOC.setCreateTime(fatDetailResponse.getCreateTime().getTime());
                goodsDOC.setUpdateTime(fatDetailResponse.getUpdateTime().getTime());
                goodsDOC.setMainImage(fatDetailResponse.getMainImage());
                Category category = goodsDocConverter.toCategoryDOC(fatDetailResponse.getCategory());
                goodsDOC.setCategory(category);
                List<GoodsDOC.Sku> skus = goodsDocConverter.toSkuDOC(fatDetailResponse.getSkus());
                goodsDOC.setSkus(skus);
                List<GoodsDOC.Param> params = goodsDocConverter.toParamDOC(fatDetailResponse.getParams());
                goodsDOC.setParams(params);
                List<GoodsDOC.Tag> tags = goodsDocConverter.toTagDOC(fatDetailResponse.getTags());
                goodsDOC.setTags(tags);
                goodsDOCRepo.save(goodsDOC);
            }
            case "DELETE" -> goodsDOCRepo.deleteById(String.valueOf(goodsEntity.getId()));
        }
        log.info("processing done！！！");
    }

    private Category category(Long categoryId) {
        CategorySummaryResponse response = categoryFeign.findOne(categoryId);

        Category category = new Category();
        category.setId(response.getId());
        category.setCategoryName(response.getCategoryName());
        category.setPath(response.getPath());
        category.setIsDisplay(response.getIsDisplay());
        category.setSort(response.getSort());
        category.setUnit(response.getUnit());
        category.setCreateTime(response.getCreateTime().getTime());
        category.setUpdateTime(response.getUpdateTime().getTime());
        return category;
    }

    @Override
    protected List<EntityComposite<GoodsBinlogEntity>> hashIndex(int concurrentlyLevel, ConsumerRecord<String, String> messageExt) {
        log.info("再次分区，concurrentlyLevel={},key={},partition={},offset={}", concurrentlyLevel, messageExt.key(), messageExt.partition(), messageExt.offset());
        String body = messageExt.value();
        TypeReference<BaseCanalBinlog<GoodsBinlogEntity>> typeReference = new TypeReference<>() {
        };
        BaseCanalBinlog<GoodsBinlogEntity> bean = JSONUtil.toBean(body, typeReference, true);
        return getEntityComposites(concurrentlyLevel, bean);
    }

    @Override
    @KafkaListener(topics = TOPIC, groupId = GROUP, containerFactory = BATCH_CONTAINER_FACTORY)
    public void onMessage(List<ConsumerRecord<String, String>> records, Acknowledgment acknowledgment) {
        super.consumeMessage(records,acknowledgment);
    }

    public interface GoodsRetryService {
        void doUpdateQuery(GoodsBinlogEntity goodsEntity, String finalScriptSource, Map<String, JsonData> finalParams) throws IOException;
    }

    @Service
    public static class GoodsRetryServiceImpl implements GoodsRetryService {
        private final ElasticsearchClient elasticsearchClient;

        public GoodsRetryServiceImpl(ElasticsearchClient elasticsearchClient) {
            this.elasticsearchClient = elasticsearchClient;
        }

        @Override
        @Retryable(retryFor = Exception.class, maxAttempts = 3, backoff = @Backoff(delay = 1000, multiplier = 1.5))
        public void doUpdateQuery(GoodsBinlogEntity goodsEntity, String finalScriptSource, Map<String, JsonData> finalParams) throws IOException {
            UpdateByQueryRequest request = UpdateByQueryRequest.of(
                    builder -> builder
                            .index("goods")
                            .query(Query.of(q ->
                                    q.term(t ->
                                            t.field("id").value(goodsEntity.getId()))
                            ))
                            .script(Script.of(s -> s.inline(
                                    new InlineScript.Builder().lang("painless").source(finalScriptSource).params(finalParams).build()
                            )))
                            .conflicts(Conflicts.Abort)
            );
            log.info("{}", request);
            elasticsearchClient.updateByQuery(request);
        }
    }
}